

namespace Stee.CAP8.MessageBus.Kafka.Subscriber
{
    using System;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.Text;
    using System.Threading;
    using Confluent.Kafka;
    using NLog;
    using Stee.CAP8.MessageBus.Interface.Delegates;
    using Stee.CAP8.MessageBus.Interface.Models;
    using Stee.CAP8.MessageBus.Interface.Utility;
    using Stee.CAP8.MessageBus.Kafka.Configurations;
    using Stee.CAP8.MessageBus.Kafka.Constants;
    using Stee.CAP8.MessageBus.Kafka.Utility;

    /// <summary>
    /// Poll consumption implementation
    /// </summary>
    public class PollingImpl : IDisposable
    {
        #region Constant & Static Variables

        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();

        /// <summary>
        /// Timer interval for re-subscription
        /// </summary>
        private const int TimerIntervalSubscribe = 5000;

        #endregion Constant & Static Variables


        #region Fields

        /// <summary>
        /// Handler for processing incoming messages
        /// </summary>
        private readonly HandleSubscribeMessage subscribeMessageHandler;

        /// <summary>
        /// Handler for processing incoming commands
        /// </summary>
        private readonly HandleSubscribeCommand subscribeCommandHandler;

        /// <summary>
        /// Handler for processing incoming events
        /// </summary>
        private readonly HandleSubscribeEvent subscribeEventHandler;

        /// <summary>
        /// Subscriber error handler
        /// </summary>
        private readonly HandleBrokerError subscribeErrorHandler;

        /// <summary>
        /// Consumer configuration
        /// </summary>
        private readonly ConsumerConfig consumerConfig;

        /// <summary>
        /// Kafka consumer builder
        /// </summary>
        private ConsumerBuilder<string, byte[]> consumerBuilder;

        /// <summary>
        /// Kafka consumer
        /// </summary>
        private IConsumer<string, byte[]> consumer;

        /// <summary>
        /// Cancel token for 
        /// </summary>
        private CancellationTokenSource cancellationToken;

        /// <summary>
        /// Flag indicating polling task is cancelled
        /// </summary>
        private bool isCancelled = false;

        /// <summary>
        /// Flag indicating polling restart
        /// </summary>
        private bool isRestartPolling = false;

        /// <summary>
        /// Timer for re-subscribing
        /// </summary>
        private readonly System.Timers.Timer resubscribeTimer;

        /// <summary>
        /// Mutex lock
        /// </summary>
        private readonly object mutex = new object();

        #endregion Fields


        #region Properties

        /// <summary>
        /// Gets the message topic
        /// Mandatory configuration.
        /// </summary>
        public TopicName TopicName { get; }

        /// <summary>
        /// Gets the client group id string.
        /// All clients sharing the same group.id belong to the same group.
        /// default: '' importance: high
        /// Mandatory configuration.
        /// </summary>
        public string GroupID
        {
            get { return (this.consumerConfig == null) ? string.Empty : this.consumerConfig.GroupId; }
        }

        /// <summary>
        /// Gets an optional identifier of a Kafka consumer(in a consumer group) that is passed to a Kafka broker with every request.
        /// The sole purpose of this is to be able to track the source of requests beyond just ip and port
        /// by allowing a logical application name to be included in Kafka logs and monitoring aggregates.
        /// </summary>
        public string ClientID { get; private set; } = string.Empty;

        /// <summary>
        /// Gets the flag indicating if consumer is currently consuming messages
        /// </summary>
        public bool IsConnected { get; private set; }

        /// <summary>
        /// Gets the flag indicating if publisher is disposed
        /// </summary>
        public bool IsDisposed { get; private set; } = false;

        #endregion Properties


        #region Constructors & Initializers

        /// <summary>
        /// Initializes a new instance of the PollingImpl class
        /// </summary>
        /// <param name="topicName">Subscribe topic name</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        private PollingImpl(TopicName topicName, HandleBrokerError brokerErrorHandler)
        {
            ValidateHelper.ValidateTopic(topicName);

            if (brokerErrorHandler == null)
            {
                throw new ArgumentNullException($"Error handler for broker is null: {nameof(brokerErrorHandler)}");
            }

            this.TopicName = topicName;
            this.subscribeErrorHandler = brokerErrorHandler;

            Dictionary<string, string> configs = ConfigurationHelper.ReadJSONConfiguration(ConfigurationHelper.MessageBusSubscribeConfig);
            SubscriberConfiguration configuration = new SubscriberConfiguration(configs);

            this.consumerConfig = new ConsumerConfig
            {
                BootstrapServers = configuration.BootstrapServers,
                GroupId = configuration.GroupID,
                ClientId = configuration.ClientID,
                AutoOffsetReset = configuration.AutoOffsetReset,
                EnableAutoCommit = configuration.EnableAutoCommit,
                EnableAutoOffsetStore = configuration.EnableAutoOffsetStore,
                MaxPollIntervalMs = configuration.MaxPollIntervalMs,
                SessionTimeoutMs = configuration.SessionTimeoutMs,
                AutoCommitIntervalMs = configuration.AutoCommitIntervalMs,
                CheckCrcs = configuration.CheckCrcs,
                PartitionAssignmentStrategy = configuration.PartitionAssignmentStrategy,
            };

            //// Create re-subscribe timer
            this.resubscribeTimer = new System.Timers.Timer(TimerIntervalSubscribe);
            this.resubscribeTimer.Elapsed += this.ResubscribeTimerOnElapsed;

            //// Initialize consumer builder
            this.InitializeConsumer();
        }

        /// <summary>
        /// Initializes a new instance of the PollingImpl class
        /// </summary>
        /// <param name="topicName">Subscribe topic name</param>
        /// <param name="subscribeMessageHandler">Subscribe message handler</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        public PollingImpl(TopicName topicName, HandleSubscribeMessage subscribeMessageHandler, HandleBrokerError brokerErrorHandler) : this(topicName, brokerErrorHandler)
        {
            this.subscribeMessageHandler = subscribeMessageHandler ?? throw new ArgumentNullException($"Message handler is not defined: {nameof(subscribeMessageHandler)}");
        }

        /// <summary>
        /// Initializes a new instance of the PollingImpl class
        /// </summary>
        /// <param name="topicName">Subscribe topic name</param>
        /// <param name="subscribeCommandHandler">Subscribe command handler</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        public PollingImpl(TopicName topicName, HandleSubscribeCommand subscribeCommandHandler, HandleBrokerError brokerErrorHandler) : this(topicName, brokerErrorHandler)
        {
            this.subscribeCommandHandler = subscribeCommandHandler ?? throw new ArgumentNullException($"Message handler is not defined: {nameof(subscribeCommandHandler)}");
        }

        /// <summary>
        /// Initializes a new instance of the PollingImpl class
        /// </summary>
        /// <param name="topicName">Subscribe topic name</param>
        /// <param name="subscribeEventHandler">Subscribe event handler</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        public PollingImpl(TopicName topicName, HandleSubscribeEvent subscribeEventHandler, HandleBrokerError brokerErrorHandler) : this(topicName, brokerErrorHandler)
        {
            this.subscribeEventHandler = subscribeEventHandler ?? throw new ArgumentNullException($"Message handler is not defined: {nameof(subscribeEventHandler)}");
        }

        /// <summary>
        /// Destructor
        /// </summary>
        ~PollingImpl()
        {
            this.Dispose();
        }


        /// <summary>
        /// Initializes the consumer builder
        /// </summary>
        private void InitializeConsumer()
        {
            //// Note: If a key or value deserializer is not set, the 
            //// deserializer corresponding to the appropriate type from Confluent.Kafka.Deserializers
            //// will be used automatically (where available). The default deserializer for string
            //// is UTF8. The default deserializer for Ignore returns null for all input data (including non-null data).
            this.consumerBuilder = new ConsumerBuilder<string, byte[]>(this.consumerConfig);
            this.consumerBuilder.SetKeyDeserializer(Deserializers.Utf8);
            this.consumerBuilder.SetValueDeserializer(Deserializers.ByteArray);
            this.consumerBuilder.SetErrorHandler(this.SubscribeErrorHandler);
            //this.consumerBuilder.SetLogHandler();

            //// Build consumer
            lock (this.mutex)
            {
                this.consumer = this.consumerBuilder.Build();
            }
        }

        #endregion Constructors & Initializers


        #region Public Methods

        /// <summary>
        /// Start poll loop to consume message
        /// </summary>
        public void StartPolling()
        {
            try
            {
                Logger.Info("Polling task started!");
#if DEBUG
                Debug.WriteLine($"[{DateTime.Now}] Polling task started!");
#endif

                this.StopSubscribeTimer();

                this.cancellationToken = new CancellationTokenSource();
                this.isCancelled = false;

                if (this.consumer != null)
                {
                    this.consumer.Subscribe(this.TopicName.Name);
                    this.IsConnected = true;

                    //// Consumer poll loop (endless)
                    while (this.IsConnected)
                    {
                        this.Consume();
                    }
                }
            }
            catch (OperationCanceledException ex)
            {
                Logger.Info(ex, "Subscription has been cancelled.");
                this.isCancelled = true;
                this.IsConnected = false;
                this.isRestartPolling = (this.cancellationToken == null) ?
                    false :
                    !this.cancellationToken.IsCancellationRequested;
            }
            catch (KafkaRetriableException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                this.IsConnected = false;
                this.isRestartPolling = true;
            }
            catch (ConsumeException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                this.IsConnected = false;
                this.isRestartPolling = true;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                this.IsConnected = false;
                this.isRestartPolling = false;
                this.StopSubscribeTimer();
                throw;
            }
            finally
            {
                if (this.isRestartPolling)
                {
                    this.RestartSubscribeTimer();
                }
                else
                {
                    this.StopSubscribeTimer();
                }
            }
        }

        /// <summary>
        /// Stop polling message
        /// </summary>
        public void StopPolling()
        {
            try
            {
                if (this.cancellationToken != null && !this.cancellationToken.IsCancellationRequested)
                {
                    this.cancellationToken.Cancel();
                    this.cancellationToken.Dispose();
                }

                this.cancellationToken = null;
                this.isCancelled = true;

                Logger.Info("Polling task stopped!");
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Commit message offset
        /// </summary>
        /// <param name="brokerMessage">Kafka cluster message, contains partition and offset information for commit</param>
        public bool Commit(object brokerMessage)
        {
            try
            {
                if (brokerMessage == null)
                {
                    throw new ArgumentNullException($"Null broker message encountered: {nameof(brokerMessage)}");
                }
                else if (!(brokerMessage is ConsumeResult<string, byte[]>))
                {
                    throw new ArgumentException($"Invalid broker message type encountered: {brokerMessage.GetType().FullName}");
                }
                else if (this.consumer == null)
                {
                    throw new ArgumentNullException("Subscriber have been disposed, create new subscriber instance required.");
                }

                ConsumeResult<string, byte[]> consumeResult = brokerMessage as ConsumeResult<string, byte[]>;
                this.consumer.Commit(consumeResult);

                return true;
            }
            catch (KafkaException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                this.SubscribeErrorHandler(this.consumer, ex.Error);
                return false;
            }
        }

        /// <summary>
        /// Dispose polling task
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion Public Methods


        #region Private Methods

        /// <summary>
        /// Consume/poll for messages
        /// </summary>
        private void Consume()
        {
            this.cancellationToken.Token.ThrowIfCancellationRequested();
            ConsumeResult<string, byte[]> consumeResult = this.consumer.Consume(this.cancellationToken.Token);

            if (consumeResult.IsPartitionEOF)
            {
                //// Reached end of topic X in partition Y at offset Z
                Logger.Info($"Reached end of topic: {consumeResult.Topic}, Partition: {consumeResult.Partition}, Offset: {consumeResult.Offset}.");
                return;
            }

            //// Extract message headers
            Dictionary<string, byte[]> properties = GetPropertiesFromHeader(consumeResult.Message.Headers);

            //// Extract header for main message type
            string messageType = null;

            if (consumeResult.Message.Headers.TryGetLastBytes(HeaderKeys.HeaderKeyMessageType, out var messageTypeHeader))
            {
                messageType = DeserializeString(messageTypeHeader, Encoding.UTF8);
            }

            //// TODO: Handle consume message success/failure state with auto commit state (default: auto commit is enabled)
            //// Handle message notification
            if (this.HandleSubscription(consumeResult, properties, messageType))
            {
                Logger.Info("Message consumed successfully.");
                this.Commit(consumeResult);
            }
            else
            {
                Logger.Warn("[Warning] Message consumption failed.");
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="consumeResult">Raw message from Kafka broker</param>
        /// <param name="properties">Message meta-data</param>
        /// <param name="messageType"></param>
        /// <returns></returns>
        private bool HandleSubscription(ConsumeResult<string, byte[]> consumeResult, Dictionary<string, byte[]> properties, string messageType)
        {
            bool isConsumed = false;

            if (string.Equals(messageType, HeaderKeys.MessageTypeCommand, StringComparison.InvariantCultureIgnoreCase))
            {
                CommandMessage commandMsg = JsonSerializer.Deserialize<CommandMessage>(consumeResult.Message.Value);
                isConsumed = this.subscribeCommandHandler(this.TopicName, commandMsg);
            }
            else if (string.Equals(messageType, HeaderKeys.MessageTypeEvent, StringComparison.InvariantCultureIgnoreCase))
            {
                EventMessage eventMsg = JsonSerializer.Deserialize<EventMessage>(consumeResult.Message.Value);
                isConsumed = this.subscribeEventHandler(this.TopicName, eventMsg);
            }
            else
            {
                isConsumed = this.subscribeMessageHandler(this.TopicName, messageType, consumeResult.Message.Value, properties);
            }

            return isConsumed;
        }

        /// <summary>
        /// Set the handler to call on error events e.g. connection failures or all brokers down.
        /// Note that the client will try to automatically recover from errors that are not marked as fatal.
        /// Non-fatal errors should be interpreted as informational rather than catastrophic.
        /// </summary>
        /// <param name="iconsumer">Consumer instance</param>
        /// <param name="error">Kafka error</param>
        private void SubscribeErrorHandler(IConsumer<string, byte[]> iconsumer, Error error)
        {
            if (error == null)
            {
                throw new ArgumentNullException($"Null Kafka error encountered: {nameof(error)}");
            }

            if (this.IsConnected && !this.isCancelled && !this.IsDisposed)
            {
                this.RestartSubscribeTimer();
            }

            this.subscribeErrorHandler?.Invoke(error.Reason, error.IsFatal);
        }

        /// <summary>
        /// Dispose consumer
        /// </summary>
        private void DisposeConsumer()
        {
            try
            {
                lock (this.mutex)
                {
                    if (this.consumer != null)
                    {
                        this.consumer.Unsubscribe();
                        this.consumer.Close();
                        this.consumer.Dispose();

                        this.consumer = null;
                        this.consumerBuilder = null;
                    }
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Dispose
        /// </summary>
        /// <param name="disposing"></param>
        protected virtual void Dispose(bool disposing)
        {
            try
            {
                if (!this.IsDisposed)
                {
                    this.DisposeConsumer();
                    this.resubscribeTimer.Stop();
                    this.resubscribeTimer.Dispose();
                }

                this.IsDisposed = true;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }


        #region Resubscribe Timer

        /// <summary>
        /// Start re-subscribe timer
        /// </summary>
        /// <returns>Return TRUE if started else FALSE</returns>
        private bool RestartSubscribeTimer()
        {
            try
            {
                this.resubscribeTimer.Stop();
                this.resubscribeTimer.Start();

                return this.resubscribeTimer.Enabled;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                return false;
            }
        }

        /// <summary>
        /// Stop re-subscribe timer
        /// </summary>
        /// <returns>Return TRUE if stopped else FALSE</returns>
        private bool StopSubscribeTimer()
        {
            try
            {
                this.resubscribeTimer.Stop();

                return !this.resubscribeTimer.Enabled;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                return false;
            }
        }

        /// <summary>
        /// Re-subscribe timer elapsed event
        /// </summary>
        /// <param name="sender">Trigger source</param>
        /// <param name="e">Event argument</param>
        private void ResubscribeTimerOnElapsed(object sender, System.Timers.ElapsedEventArgs e)
        {
            try
            {
                //// Log timer elapsed event
                Logger.Info("Restarting polling task...");
#if DEBUG
                Debug.WriteLine($"[{DateTime.Now}] Restarting polling task...");
#endif

                this.StopPolling();
                this.StartPolling();
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
            }
        }

        #endregion Resubscribe Timer

        #endregion Private Methods


        #region Static Methods

        /// <summary>
        /// Get properties from Kafka message headers
        /// </summary>
        /// <param name="headers">Kafka message headers</param>
        /// <returns>Return property collection mapping if available else NULL</returns>
        private static Dictionary<string, byte[]> GetPropertiesFromHeader(Headers headers)
        {
            Dictionary<string, byte[]> properties = null;

            if (headers != null)
            {
                properties = new Dictionary<string, byte[]>();
                var headerEnumtr = headers.GetEnumerator();

                while (headerEnumtr.MoveNext())
                {
                    var iHeader = headerEnumtr.Current;

                    //// Get header key
                    var key = iHeader.Key;

                    //// Get properties where key is NOT message type
                    if (!key.Equals(HeaderKeys.HeaderKeyMessageType, StringComparison.InvariantCultureIgnoreCase))
                    {
                        //// Get header value
                        properties[key] = iHeader.GetValueBytes();
                    }
                }
            }

            return properties;
        }

        /// <summary>
        /// De-serialize byte array in to string
        /// </summary>
        /// <param name="bytes">Byte array to de-serialize</param>
        /// <param name="encoding">Encoding type</param>
        /// <returns>Return deserialized byte arrary as string</returns>
        private static string DeserializeString(byte[] bytes, Encoding encoding)
        {
            return (bytes == null || encoding == null) ? null : encoding.GetString(bytes);
        }

        #endregion Static Methods
    }
}
